package elasticsearch

import (
	"context"
	"fmt"
	"github.com/olivere/elastic/v7"
	"strings"
	"time"
)

// 初始化es,接收从kafka传递的数据
var (
	client *elastic.Client
	ch     chan *LogData
)

// 日志结构
type LogData struct {
	Topic string `json:"topic"`
	Data  string `json:"data"`
}

func InitES(address string) (err error) {
	// 判断是否存在 http:// 前缀
	if !strings.HasPrefix(address, "http://") {
		address = "http://" + address
	}
	// 获取es client 设置不通过docker 容器获取
	client, err = elastic.NewClient(elastic.SetURL(address), elastic.SetSniff(false))
	if err != nil {
		panic("create es client err")
		return err
	}
	fmt.Println("es init client success")
	// 初始化es chancel
	ch = make(chan *LogData)
	// 将数据发送到es中
	go sendToES()
	return
}

func sendToES() {
	for {
		select {
		case msg := <-ch:
			_, err := client.Index().Index(msg.Topic).Type("web_log").BodyJson(msg).Do(context.Background())
			if err != nil {
				fmt.Println(err)
				continue
			}
			//fmt.Printf("Indexed student %v to index %s, type %s\n", do.Id, do.Index, do.Type)
		default:
			// 睡一会
			time.Sleep(time.Second)
		}

	}
}

// SendToES 发送数据到ES
func SendToESChan(msg *LogData)  {
	ch <- msg
}
